package com.sui.bigdata.flink.sql.function.aggregate;

import com.sui.bigdata.flink.sql.function.util.NameUnusualUtil;

import org.apache.flink.table.functions.AggregateFunction;

import java.math.BigDecimal;
import java.util.Iterator;
import java.util.Map;
import java.util.Set;

/**
 * @author LiuHuan
 * @date 2019/9/5 14:18
 * @description
 * @reviewer
 */
//通讯录异常联系人姓名占比
public class UserknNameUnusualRate extends AggregateFunction<Double, NameUnusualUtil> {
    @Override
    public NameUnusualUtil createAccumulator() {
        return new NameUnusualUtil();
    }

    @Override
    public Double getValue(NameUnusualUtil nameUnusualUtil) {
        Double result = 0.0000;
        Integer count = 0;
        if (!(nameUnusualUtil == null || nameUnusualUtil.getCount() == 0 || nameUnusualUtil.getGroups().isEmpty())) {
            Set<Map.Entry<String, Integer>> entries = nameUnusualUtil.getGroups().entrySet();
            for (Map.Entry<String, Integer> entry : entries
                    ) {
                count += entry.getValue();
            }
        }
        if (nameUnusualUtil.getCount() > 0) {
            Double newResult = (double) count / nameUnusualUtil.getCount();
            if (newResult != null && newResult >= 0) {
                BigDecimal bigDecimal = new BigDecimal(newResult);
                result = bigDecimal.setScale(4, BigDecimal.ROUND_HALF_UP).doubleValue();
            }
        }
        return result;
    }

    public void accumulate(NameUnusualUtil nameUnusualUtil, String name) {
        if (!nameUnusualUtil.exitsName(name)) {
            //异常num为1，否则为0
            Integer num;
            nameUnusualUtil.addName(name);
            nameUnusualUtil.setCount(nameUnusualUtil.getCount() + 1);
            if (isFigureName(name) || name.length() > 5 || isEnglishName(name)|| isSpecialName(name)){
                num = 1;
            }else {
                num = 0;
            }
            nameUnusualUtil.getGroups().put(name,num);
        }
    }
    public  boolean isEnglishName(String name){

        boolean isWord = name.matches("[a-zA-Z]+");
        return isWord;
    }
    public  boolean isFigureName(String name){
        boolean isFigure = name.matches("^[-\\+]?[\\d]*$");
        return isFigure;
    }
    public  boolean isSpecialName(String name){
        boolean isSpecial = !name.matches(".*[0-9a-zA-Z].*|.*[\\u4e00-\\u9fa5].*");
        return isSpecial;
    }

    public void merge(NameUnusualUtil nameUnusualUtil , Iterable<NameUnusualUtil> it) {
        Iterator<NameUnusualUtil> iter = it.iterator();
        while (iter.hasNext()) {
            NameUnusualUtil a = iter.next();
            nameUnusualUtil.setGroups(a.getGroups());
            nameUnusualUtil.setCount(a.getCount());
            nameUnusualUtil.setRepetition(a.getRepetition());
        }
    }
}
